package com.you.learning.rocketmq.producer;

import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.concurrent.*;

/**
 * @Description 带事务机制的消息发送
 * @Author yough
 * @Version V1.0.0
 * @Since 1.0
 * @Date 2021/3/26
 */
public class TransactionProducer {
    public static void main(String[] args) throws Exception {
        // 创建一个支持事务消息的Producer
        TransactionMQProducer producer = new TransactionMQProducer("test_transaction_producer");
        // 创建一个线程池，用来处理RocketMQ回调你的请求
        ExecutorService executorService = new ThreadPoolExecutor(
                2,
                5,
                100,
                TimeUnit.SECONDS,
                new ArrayBlockingQueue<>(2000),
                new ThreadFactory() {
                    @Override
                    public Thread newThread(Runnable r) {
                        Thread t = new Thread(r);
                        t.setName("TestThread");
                        return t;
                    }
                });
        // 给事务消息生产者设置线程池，负责执行RocketMQ的回调请求
        producer.setExecutorService(executorService);
        TransactionListener transactionListener = new TransactionListenerImpl();
        // 设置Rocket事务消息的回调函数
        producer.setTransactionListener(transactionListener);
        // 启动生产者
        producer.start();
        // 发送消息
        Message msg = new Message("PayOrderSuccessTopic", "TestTag", "testKey", "订单支付消息".getBytes());
        try {
            // 发送事务消息
            TransactionSendResult sendResult = producer.sendMessageInTransaction(msg, null);
        } catch (Exception ex) {
            // half消息发送失败，订单系统执行回滚逻辑，比如触发支付退款，更新订单状态为"已关闭"
        }
    }

    static class TransactionListenerImpl implements TransactionListener {
        // 如果发送half消息成功，RocketMQ会在这里回调你的这个函数，你就可以执行本地事务了
        @Override
        public LocalTransactionState executeLocalTransaction(Message message, Object o) {
            // 执行本地事务
            // 然后根据本地一连串事务执行结果，去选择执行commit or rollback
            try {
                // 执行本地事务
                // 执行完本地事务后，返回commit标记给RocketMQ，告诉RocketMQ提交half消息
                return LocalTransactionState.COMMIT_MESSAGE;
            } catch (Exception ex) {
                // 如果本地事务执行失败，回滚之前一切所做的所有操作
                // 同时返回rollback,告诉RocketMQ，标记half消息无效
                return LocalTransactionState.ROLLBACK_MESSAGE;
            }
        }

        // 如果因为各种原因，RocketMQ没有收到commit或rollback，在一定的时间内，RocketMQ会回调这个函数
        @Override
        public LocalTransactionState checkLocalTransaction(MessageExt msg) {
            // 查询本地事务，是否执行成功了
            Integer localStatus = 0;
            //Integer localStatus = localTrans.get(msg.getTransactionId());
            if (null != localStatus) {
                switch (localStatus) {
                    case 0:
                        return LocalTransactionState.UNKNOW;
                    case 1:
                        return LocalTransactionState.COMMIT_MESSAGE;
                    case 2:
                        return LocalTransactionState.ROLLBACK_MESSAGE;
                }
            }
            return LocalTransactionState.COMMIT_MESSAGE;
        }
    }
}
